package com.hucais.app.ods;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hucais.bean.monitor.SelectMonitorDetail;
import com.hucais.utils.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 重点品监测
 */
public class SelectMonitorApp {
    private static final Logger LOG = LoggerFactory.getLogger(SelectMonitorApp.class);
    private static final String SOURCE_TOPIC = DefaultPropertiesUtil.get("kafka.topic.monitor");
    private static final String GROUP_ID = "select_monitor_app_group";
    private static final String CHECK_POINT_PATH_SUF = DefaultPropertiesUtil.get("checkpoint.path.suf.monitor");

    public static void main(String[] args) {
        //获取执行环境
        StreamExecutionEnvironment env = AppCommonUtil.getDefaultStreamEnv(false, CHECK_POINT_PATH_SUF, 1);

        try {
            DataStreamSource<String> kafkaDS = env.addSource(KafkaUtil.getDefaultFlinkKafkaConsumer(SOURCE_TOPIC, GROUP_ID));

            // 将每行数据转换为JSON对象
            SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);
            //过滤空值+主键缺失
            SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS
                    .filter(jsonObj -> jsonObj.size() > 0)
                    .filter(jsonObj -> StringUtils.isNotBlank(jsonObj.getString("task_code")) && StringUtils.isNotBlank(jsonObj.getString("channel")) &&
                            StringUtils.isNotBlank(jsonObj.getString("acq_date")) && StringUtils.isNotBlank(jsonObj.getString("acq_time")) &&
                            StringUtils.isNotBlank(jsonObj.getString("isbn")) && StringUtils.isNotBlank(jsonObj.getString("uuid")));

            // 组装Sink对象流
            SingleOutputStreamOperator<SelectMonitorDetail> detailDS = filterDS.map(jsonObj -> {
                SelectMonitorDetail detail = new SelectMonitorDetail();
                //提取采集小时
                String acqTime = CommonUtil.handleBlankStr(jsonObj.getString("acq_time"));
                DateTime dateTime = DateUtil.parseDateTime(acqTime);
                int acqHour = dateTime.hour(true);

                detail.setTask_code(jsonObj.getString("task_code"));
                detail.setAcq_date(jsonObj.getString("acq_date"));
                detail.setAcq_hour(acqHour);
                detail.setChannel(jsonObj.getString("channel"));
                detail.setUuid(jsonObj.getString("uuid"));
                detail.setIsbn(jsonObj.getString("isbn"));
                detail.setProduct_status(jsonObj.getShort("product_status"));
                detail.setBook_name(CommonUtil.handleBlankStr(jsonObj.getString("book_name")));
                detail.setAuthor(CommonUtil.handleBlankStr(jsonObj.getString("author")));
                detail.setPublishing_house(CommonUtil.handleBlankStr(jsonObj.getString("publishing_house")));
                detail.setPublishing_time(CommonUtil.handleBlankStr(jsonObj.getString("publishing_time")));
                detail.setBrand(CommonUtil.handleBlankStr(jsonObj.getString("brand")));
                detail.setStore_name(CommonUtil.handleBlankStr(jsonObj.getString("store_name")));
                detail.setStore_pricing(CommonUtil.handleBlankDouble(jsonObj.getString("store_pricing")));
                detail.setSelling_price(CommonUtil.handleBlankDouble(jsonObj.getString("selling_price")));
                detail.setAcq_url(CommonUtil.handleBlankStr(jsonObj.getString("acq_url")));
                detail.setAcq_time(acqTime);
                detail.setComments_cnt(CommonUtil.handleBlankNumber(jsonObj.getString("comments_cnt")));
                return detail;
            });
            // 数据输出
            String sqlTemplate = "INSERT INTO %s (task_code, acq_date,acq_hour,channel, uuid, isbn, product_status, book_name, author, publishing_house, publishing_time, brand, store_name, store_pricing, selling_price, acq_url, acq_time, comments_cnt) " +
                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "isbn = VALUES(isbn),product_status = VALUES(product_status),book_name = VALUES(book_name), " +
                    "author = VALUES(author),publishing_house = VALUES(publishing_house),publishing_time = VALUES(publishing_time), " +
                    "brand = VALUES(brand),store_name = VALUES(store_name),store_pricing = VALUES(store_pricing), " +
                    "selling_price = VALUES(selling_price),acq_url = VALUES(acq_url),acq_time = VALUES(acq_time),comments_cnt = VALUES(comments_cnt)";
            String sql = String.format(sqlTemplate, DefaultPropertiesUtil.get("mysql.table.monitor"));
            detailDS.addSink(JdbcUtil.getMysqlSinkFunc(sql));
            // 执行
            env.execute("SelectMonitorApp");
        } catch (Exception e) {
            LOG.error("重点品监测-出现异常", e);
        }
    }
}
